package SQL_L

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.junit.Test

class ReaderWriter {
  @Test
  def Reader1(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("reader1")
      .getOrCreate()
    //框架
    //第一种读取形式
    spark.read
      .format("csv")
      .option("header", true)
      .option("inferSchema", true) //文件的数据类型（每一行）
      .load("data/BeijingPM20100101_20151231.csv")
      .show()
    //第二种读取形式
    spark.read
      .option("header", true)
      .option("inferSchema", true) //文件的数据类型（每一行）
      .csv("data/BeijingPM20100101_20151231.csv")
      .show()
  }

  @Test
  def writer1(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("writer1")
      .getOrCreate()
    //读取数据集
    val df = spark.read.option("header", true).csv("data/BeijingPM20100101_20151231.csv")
    //写入数据集
    df.write.json("data/Beijing_pm2.json")
    df.write.format("json").save("data/Beijing_pm2.json")
  }

  @Test
  def parquetTest(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("parquet")
      .getOrCreate()
    //读取csv文件的数据
    val df = spark.read.option("header", true).csv("data/BeijingPM20100101_20151231.csv")
    //将数据写为parquet格式
    df.write
      //mode时写入格式，有四种：报错、覆盖，追加和忽略
      .mode(SaveMode.Overwrite) //存储位置已有文件则重写
      .format("parquet")
      .save("data/beijing_pm3")
    //读取parquet文件格式
  }

  /**
   * 表分区的概念不仅仅在parquet上有。其他格式的文件也可以指定表分区
   */
  @Test
  def parquetPartitions(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("parquetPartition")
      .getOrCreate()
    //读取数据
    val df = spark.read
      .option("header", true)
      .csv("data/BeijingPM20100101_20151231.csv")
    //写文件，表分区
//    df.write
//      .partitionBy("year", "month")
//      .save("data/beijing_pm4")
    //读文件，自动发现分区
    spark.read
      .parquet("data/beijing_pm4/year=2010/month=1")
      .show()
  }

  /**
   * 关于json的一些练习
   */
  @Test
  def jsonTest(): Unit = {
    //创建SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("jsonTest")
      .getOrCreate()
    //读取数据
    val df = spark.read
      .option("header", true)
      .csv("data/BeijingPM20100101_20151231.csv")
    //    df.write
    //      .json("data/beijing_pm.json")
    //直接将文件格式转换成json（一般对象都是转换成json）
    df.toJSON
    //把RDD转为DataFrame
    val jsonRDD = df.toJSON.rdd
    spark.read
      .json("data/beijing_pm.json")
      .show()
  }
}
